package lib
//https://github.com/confluentinc/confluent-kafka-go/
import (
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"app/conf"
)
var p  *kafka.Producer
func InitKafka() {
	var err error
	p, err = kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": conf.KFK_SERVER})
	if err != nil {
		panic(err)
	}

	// Delivery report handler for produced messages
	go func() {
		for e := range p.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
				} else {
					fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
				}
			}
		}
	}()
}
func ProduceKafkaMsg(topic, msg string)(string,error){
	//初始化kafka
	InitKafka()
	defer p.Close()
	err := p.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value:          []byte(msg),
	}, nil)
	// Wait for message deliveries
	p.Flush(15 * 1000)
	return msg,err
}